package com.rem.flink.flink10Sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 时间属性和窗口
 * <p>
 * 窗口表值函数
 * * 滚动窗口（Tumbling Windows）；
 * * 滑动窗口（Hop Windows，跳跃窗口）；
 * * 累积窗口（Cumulate Windows）；
 *
 * @author Rem
 * @date 2022-11-07
 */

public class TimeAndWindowTest2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 在创建表的DDL中直接定义时间属性
        String createDdl = "CREATE TABLE clickTable (" +
                " user_name STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " et AS TO_TIMESTAMP( FROM_UNIXTIME(ts / 1000) ), " +  //BIGINT类型转为时间戳类型
                " WATERMARK FOR et AS et - INTERVAL '1' SECOND " +    //定义1s延迟时间的水位线
                ") WITH (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'input/clicks.csv', " +
                " 'format' =  'csv' " +
                ")";

        tableEnv.executeSql(createDdl);

        /**
         * 滚动窗口 TUMBLE  10s 一次
         */
        Table tumbleWindowResultTable = tableEnv.sqlQuery("select user_name, count(url) as cnt, " +
                "window_end as endT " +
                "from table( " +
                " TUMBLE( TABLE clickTable, DESCRIPTOR(et), INTERVAL '10' SECOND )" +
                ") " +
                "group by user_name, window_start, window_end "
        );
         // tableEnv.toDataStream(tumbleWindowResultTable).print("tumbleWindowResultTable: ");

        /**
         * 滑动窗口 HOP  10s的窗口 5s 滑一次
         */
        Table hopWindowResultTable = tableEnv.sqlQuery("select user_name, count(url) as cnt, " +
                "window_end as endT " +
                "from table( " +
                " HOP( TABLE clickTable, DESCRIPTOR(et), INTERVAL '5' SECOND,  INTERVAL '10' SECOND )" +
                ") " +
                "group by user_name, window_start, window_end "
        );
       //  tableEnv.toDataStream(hopWindowResultTable).print("hopWindowResultTable: ");

        /**
         * 累计窗口 CUMULATE 统计10s内 每5s累计一次的数据
         */
        Table cumulateWindowResultTable = tableEnv.sqlQuery("select user_name, count(url) as cnt, " +
                "window_end as endT " +
                "from table( " +
                " CUMULATE( TABLE clickTable, DESCRIPTOR(et), INTERVAL '5' SECOND,  INTERVAL '10' SECOND )" +
                ") " +
                "group by user_name, window_start, window_end "
        );
        //  tableEnv.toDataStream(cumulateWindowResultTable).print("cumulateWindowResultTable: ");


        /**
         * 累积窗口
         * 开窗聚合 OVER 针对每一行计算一个聚合值
         * 统计当前行前三行数据 ts平均值
         */
        Table overWindowResultTable = tableEnv.sqlQuery("SELECT user_name, " +
                " avg(ts) OVER (" +
                "   PARTITION BY user_name " +
                "   ORDER BY et " +
                "   ROWS BETWEEN 3 PRECEDING AND CURRENT ROW" +  //行间隔 当前行前三行
                //"   RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW"+   //范围间隔 当前数据前1个小时
                ") AS avg_ts " +
                "FROM clickTable");
        tableEnv.toDataStream(overWindowResultTable).print("overWindowResultTable: ");


        env.execute();
    }
}
